package com.ruoyi.system.utils;

import com.alibaba.cloud.commons.lang.StringUtils;
import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;
import org.apache.kafka.common.security.JaasUtils;


import java.util.Properties;

public class KafkaUtils {
    /**
     * 创建topic
     *
     * @param bootstrapServers zk的地址
     * @param topicName        topic的名称
     * @param partitions       分区数
     * @param replication      副本数
     */
    public static boolean createTopic(String bootstrapServers, String topicName, int partitions, int replication) {
        ZkUtils zkUtils = null;
        try {
            zkUtils = ZkUtils.apply(bootstrapServers, 10000, 10000, JaasUtils.isZkSaslEnabled());
            // 创建一个指定分区数量和副本数量 的topic
            if (StringUtils.isEmpty(bootstrapServers) || partitions < 1 || replication < 1) {
                return false;
            }
            if (!AdminUtils.topicExists(zkUtils, topicName)) {
                AdminUtils.createTopic(zkUtils, topicName, partitions, replication, new Properties(), RackAwareMode.Enforced$.MODULE$);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (zkUtils != null) {
                zkUtils.close();
            }
        }
        return false;
    }

    /**
     * topic的删除
     *
     * @param bootstrapServers zk地址
     * @param topicName        topic名称
     * @return
     */
    public static boolean deleteTopic(String bootstrapServers, String topicName) {
        ZkUtils zkUtils = null;
        try {
            zkUtils = ZkUtils.apply(bootstrapServers, 30000, 30000, JaasUtils.isZkSaslEnabled());
            // 创建一个指定分区数量和副本数量 的topic
            if (StringUtils.isEmpty(bootstrapServers)) {
                return false;
            }
            AdminUtils.deleteTopic(zkUtils, topicName);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (zkUtils != null) {
                zkUtils.close();
            }
        }
        return false;
    }
}
